package com.atguigu.flink.streamapi.trans;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * Created by Smexy on 2023/1/13
 *
 *
 *      卖牛肉馅。
 *
 *      猪肉9 ----> 加工为猪肉馅 ---->加牛肉味香精 -----> 牛肉味肉馅 --
 *                                                                  ----> 牛肉馅
 *      牛肉1 ----> 加工为牛肉馅 --------------------->牛肉馅 -----
 *
 *      connect可以把多个流联合在一个函数中处理，但是这个函数依旧拥有两个方法，各处理各的！最终输出必须是同一种类型！
 */
public class Demo5_Connect
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<Pork> ds1 = env.fromElements(
            new Pork("大块猪肉"),
            new Pork("小块猪肉"),
            new Pork("中块猪肉"));

        DataStreamSource<Beef> ds2 = env.fromElements( new Beef("小块牛肉"));

        ConnectedStreams<Pork, Beef> ds3 = ds1.connect(ds2);

        SingleOutputStreamOperator<String> ds4 = ds3.map(new CoMapFunction<Pork, Beef, String>()
        {
            //处理 ds1中的数据，最终输出必须是同一种类型
            @Override
            public String map1(Pork value) throws Exception {
                System.out.println("正在将:" + value + " 添加牛肉香精");
                p = value;
                return "牛肉味肉馅";
            }

            private Pork p;

            //处理 ds2中的数据，最终输出必须是同一种类型
            @Override
            public String map2(Beef value) throws Exception {
                System.out.println("正在将:" + value + " 添加膨松剂");
                System.out.println(p);
                return "牛肉味肉馅";
            }
        });

        ds4.print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Beef{
        private String type ;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Pork{
        private String type;
    }
}
